package com.atguigu.flink.datastreamapi.multistream;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2023/4/4
 *
 *
 *  把不同类型的流，两两联合，输出同一种类型。
 *
 *  国民党                      国军
 *        -----> 合作 ------>
 *  共产党                      国军
 *
 *  -----------------------------------
 *
 *
 *   猪肉 9 ---                   ----对猪肉加工，添加牛肉香精，添加防腐剂 ---搅碎
 *             ----机器(算子，加工)                                              --------->牛肉馅
 *   牛肉 1 ---                   ----对牛肉加工，添加防腐剂----搅碎
 *
 *
 *
 *
 */
public class Demo2_Connect
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        SingleOutputStreamOperator<Beef> ds1 = env.socketTextStream("hadoop102", 8888).map(x -> new Beef(x));
        SingleOutputStreamOperator<Pork> ds2 = env.socketTextStream("hadoop102", 8889).map(x -> new Pork(x));

        ConnectedStreams<Beef, Pork> ds3 = ds1.connect(ds2);

        ds3.process(new CoProcessFunction<Beef, Pork, String>()
        {
            //只处理 Beef value
            @Override
            public void processElement1(Beef value, Context ctx, Collector<String> out) throws Exception {

                System.out.println("对" + value + "添加防腐剂");
                out.collect("牛肉馅");
            }

            //只处理Pork value
            @Override
            public void processElement2(Pork value, Context ctx, Collector<String> out) throws Exception {
                System.out.println("对" + value + "添加牛肉香精，添加防腐剂");
                out.collect("牛肉馅");
            }
        })
           .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class  Beef{
        private String name ;
    }
    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class Pork{
        private String name;
    }
}
